package com.wyh.apitest.sink

import com.wyh.apitest.source.SensorReading
import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.scala._

object FileSinkDemo {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")
    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    dataStream.print()
    //直接解析出来以逗号分割
//    dataStream.writeAsCsv("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor_out.txt")

    //forRowFormat
    dataStream.addSink(StreamingFileSink.forRowFormat(
      new Path("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor_out2.txt"),
      new SimpleStringEncoder[SensorReading]()).build()
    )


    streamEnv.execute("Flie Sink")
  }

}